package com.example.xj.storm.spouts;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by Administrator on 2018/9/6.
 * 数据源类
 */
public class HelloWorldSpout extends BaseRichSpout {

    private SpoutOutputCollector spoutOutputCollector;

    private int count=0;

    private Map<String,Object> parameterMap=new HashMap<String,Object>();

    private String field="parameter";



    /**
     * Storm配置的Map;
     * topology中组件的信息;
     * 发射tuple的方法;
     * 主要用于初始化传入参数的处理
     * @param map
     * @param topologyContext
     * @param spoutOutputCollector
     */
    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        System.out.println("open......");
        parameterMap=map;
        this.spoutOutputCollector = spoutOutputCollector;
    }


    /**
     *主要执行方法，用于输出信息,通过collector.emit方法发射
     */
    @Override
    public void nextTuple() {
        if(count<2){
            System.out.println("第"+count+"次开始发送数据...");
            this.spoutOutputCollector.emit(new Values(parameterMap.get("xiongjie")));
        }
        count++;
    }

    /**
     * 用于声明数据格式。
     * 即输出的一个Tuple中，包含几个字段。
     * 因为这里我们只发射一个，所以就指定一个。如果是多个，则用逗号隔开。
     * @param outputFieldsDeclarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        System.out.println("定义格式...");
        outputFieldsDeclarer.declare(new Fields(field));
    }

    /**
     * 表示Tuple处理成功。
     * @param msgId
     */
    @Override
    public void ack(Object msgId) {
        System.out.println("ack:成功！"+msgId);
    }

    /**
     * 表示Tuple处理失败
     * @param msgId
     */
    @Override
    public void fail(Object msgId) {
        System.out.println("失败:"+msgId);
    }

    /**
     * 表示Topology停止
     */
    @Override
    public void close() {
        System.out.println("关闭...");
    }

}
